//
// Created by suzhen on 16-11-11.
//

#include "OkRedisCore.h"

using namespace NoSQL;
USING_ORION_V2

OkRedisCore::~OkRedisCore() {
    //empty
}

OkRedisCore::OkRedisCore(const std::string &ip, int port)
        : runInThreadId(),
          workQueue_(),
          ipAddr_(ip),
          port_(port),
          redisAsyncCtx(nullptr),
          commandLock_(),
          isClosing_(false),
          closePromise_(),
          startPromise_(),
          openInHandlers_(0),
          refCount_(0) {
    uv_loop_init(&okRedisEvb_);
    //
    ++openInHandlers_;
    asyncRedisCmdContext_.data = this;
    uv_async_init(&okRedisEvb_, &asyncRedisCmdContext_, &OkRedisCore::OkRedisAsyncOperateCallBack);
    //
    ++openInHandlers_;
    asyncCloseContext_.data = this;
    uv_async_init(&okRedisEvb_, &asyncCloseContext_, &OkRedisCore::OkRedisAsyncCloseCallBack);
    retain();
}

void OkRedisCore::RedisAsyncCmdCallBack(redisAsyncContext *, void *reply, void *privdata) {
    RedisCommandPackage *wp = (RedisCommandPackage *) privdata;
    redisReply *r = (redisReply *) reply;
    RawRedisReply rw(r);
    wp->invoke(rw);
    delete wp;
}

void OkRedisCore::RedisDisconnected(const redisAsyncContext *ctx, int status) {
    LOG(INFO) << "redis disconnected!!!";
    SYS_FAULT(status == REDIS_OK) << "RedisDisconnected ? Info : \t" << ctx->errstr;
    OkRedisCore *okRedisIns = (OkRedisCore *) ctx->data;
    if (!okRedisIns) {
        return;
    }
    if (okRedisIns->redisAsyncCtx) {
        //unnecessary , maybe ?
        //redisAsyncFree(okRedisIns->redisAsyncCtx);
        okRedisIns->redisAsyncCtx = NULL;
    }
    //close asyncRedisCmdContext_
    uv_close((uv_handle_t *) &okRedisIns->asyncRedisCmdContext_, &OkRedisCore::OkRedisUninstallHandlersCallBack);
    //close asyncCloseContext_
    uv_close((uv_handle_t *) &okRedisIns->asyncCloseContext_, &OkRedisCore::OkRedisUninstallHandlersCallBack);
}

void OkRedisCore::RedisConnected(const redisAsyncContext *ctx, int status) {
    LOG(INFO) << "redis connected!!!";
    SYS_FAULT(status == REDIS_OK) << "RedisDisconnected ? Info : \t" << ctx->errstr;
    OkRedisCore *okRedisIns = (OkRedisCore *) ctx->data;
    if (!okRedisIns) {
        return;
    }
    okRedisIns->startPromise_.setValue(true);
}

Future<bool> OkRedisCore::shutdown() {
    std::lock_guard<std::recursive_mutex> g(commandLock_);
    Future<bool> f = closePromise_.getFuture();
    if (isClosing_) {
        return f;
    }
    isClosing_ = true;
    if (std::this_thread::get_id() == runInThreadId) {
        OkRedisCore::OkRedisAsyncCloseCallBack(&asyncCloseContext_);
    } else {
        uv_async_send(&asyncCloseContext_);
    }
    return f;
}

Future<bool> OkRedisCore::start() {
    redisAsyncCtx = redisAsyncConnect(ipAddr_.c_str(), port_);
    redisAsyncCtx->data = this;
    SYS_FAULT(redisAsyncCtx && !redisAsyncCtx->err) << "OkRedisCore Connect Error:\t" << redisAsyncCtx->errstr;
    redisLibuvAttach(redisAsyncCtx, &okRedisEvb_);
    redisAsyncSetConnectCallback(redisAsyncCtx, &OkRedisCore::RedisConnected);
    redisAsyncSetDisconnectCallback(redisAsyncCtx, &OkRedisCore::RedisDisconnected);
    std::thread okRedisRunningThread([this]() {
        this->runInThreadId = std::this_thread::get_id();
        uv_run(&okRedisEvb_, UV_RUN_DEFAULT);
        LOG(INFO) << "uv main loop stopped!";
        for (;;) {
            int e = uv_loop_close(&okRedisEvb_);
            if (e == UV_EBUSY) {
                continue;
            }
            break;
        }
    });
    okRedisRunningThread.detach();
    return startPromise_.getFuture();
}

void OkRedisCore::executeAsyncCmd(const std::string &cmd,
                                  CallBackType &&cb) {
    SYS_FAULT(startPromise_.isFulfilled()) << "OkRedis Not Start yet";
    //
    std::lock_guard<std::recursive_mutex> g(commandLock_);
    if (isClosing_ ||
        uv_is_closing((uv_handle_t *) &asyncRedisCmdContext_) ||
        closePromise_.isFulfilled()) {
        return;
    }
    //
    SYS_FAULT(!closePromise_.isFulfilled()) << "OkRedis Already Closed";
    RedisCommandPackage *wrapper = new RedisCommandPackage(cmd, cb);
    workQueue_.push_back(wrapper);
    //
    if (std::this_thread::get_id() == runInThreadId) {
        OkRedisAsyncOperateCallBack(&asyncRedisCmdContext_);
    } else {
        uv_async_send(&asyncRedisCmdContext_);
    }
}

void OkRedisCore::OkRedisAsyncOperateCallBack(uv_async_t *async) {
    OkRedisCore *okRedisIns = (OkRedisCore *) async->data;
    std::lock_guard<std::recursive_mutex> g(okRedisIns->commandLock_);
    std::list<RedisCommandPackage *> &queue = okRedisIns->workQueue_;
    //将 RedisCommandPackage 从队列取出
    for (; !queue.empty();) {
        RedisCommandPackage *wrapper = queue.front();
        redisAsyncCommand(okRedisIns->redisAsyncCtx,
                          &OkRedisCore::RedisAsyncCmdCallBack,
                          wrapper, wrapper->getCmd().c_str());
        queue.pop_front();
    }
}

void OkRedisCore::OkRedisAsyncCloseCallBack(uv_async_t *async) {
    OkRedisCore *okRedisIns = (OkRedisCore *) async->data;
    if (okRedisIns->redisAsyncCtx) {
        //need thread safe operation
        redisAsyncDisconnect(okRedisIns->redisAsyncCtx);
    }
}


void OkRedisCore::OkRedisUninstallHandlersCallBack(uv_handle_t *hd) {
    OkRedisCore *okRedisIns = (OkRedisCore *) hd->data;
    if (--okRedisIns->openInHandlers_ == 0) {
        std::lock_guard<std::recursive_mutex> g(okRedisIns->commandLock_);
        std::list<RedisCommandPackage *> &queue = okRedisIns->workQueue_;
        for (; !queue.empty();) {
            RedisCommandPackage *wrapper = queue.front();
            delete wrapper;
            queue.pop_front();
        }
        //close loop
        uv_stop(&okRedisIns->okRedisEvb_);
        //free loop
        okRedisIns->closePromise_.setValue(true);
        okRedisIns->release();
    }
}

Future<bool> OkRedisCore::closeFuture() {
    return closePromise_.getFuture();
}

bool OkRedisCore::isClosed() {
    return closePromise_.isFulfilled();
}

void OkRedisCore::retain() {
    std::lock_guard<std::recursive_mutex> g(commandLock_);
    ++refCount_;
}

void OkRedisCore::release() {
    assert(refCount_ != 0);
    commandLock_.lock();
    --refCount_;
    commandLock_.unlock();
    if (refCount_ == 0) {
        delete this;
    }
}

OkRedisCore *OkRedisCore::createRedisCore(const std::string &ip, int port) {
    return new OkRedisCore(ip, port);
}
